package com.rex.saas.mqtt;


import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.net.ssl.KeyManager;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.TrustManagerFactory;
import java.io.InputStream;
import java.security.KeyStore;
import java.security.cert.CertificateFactory;
import java.security.cert.X509Certificate;
import org.springframework.core.io.Resource;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
/**
* @author wt
 * MQTT工具类操作
 */
@Getter
@Slf4j
@Component
public class MaMqttConnect {
    private MqttClient mqttClient;

    @Value("${mqtt.tls.ca-cert}")
    private Resource caCertResource;

    @Value("${maMqtt.broker-url}")
    private String brokerUrl;
    /**
     * 客户端connect连接mqtt服务器
     *
     * @param userName     用户名
     * @param passWord     密码
     * @param mqttCallback 回调函数
     **/
    public void setMqttClient(String clientId, String userName, String passWord, MqttCallback mqttCallback) throws Exception {
        MqttConnectOptions options = mqttConnectOptions(clientId, userName, passWord);
        mqttClient.setCallback(mqttCallback);
        mqttClient.connect(options);
    }

    /**
     * MQTT连接参数设置
     */
    private MqttConnectOptions mqttConnectOptions(String clientId, String userName, String passWord) throws Exception {
        mqttClient = new MqttClient(brokerUrl, clientId, new MemoryPersistence());
        MqttConnectOptions options = new MqttConnectOptions();
        //用户名
        options.setUserName(userName);
        //密码
        options.setPassword(passWord.toCharArray());
        //连接超时，默认：30
        options.setConnectionTimeout(60);
        //自动重新连接，默认：false
        options.setAutomaticReconnect(true);
        //清理会话，默认：true
        options.setCleanSession(false);
        //保持活动间隔，默认：60
        options.setKeepAliveInterval(60);

//        options.setSocketFactory(createSslSocketFactory());
        return options;
    }
    private SSLSocketFactory createSslSocketFactory() throws Exception {
        // 1. 加载 CA 证书
        CertificateFactory cf = CertificateFactory.getInstance("X.509");
        try (InputStream caInput = caCertResource.getInputStream()) {
            X509Certificate caCert = (X509Certificate) cf.generateCertificate(caInput);

            KeyStore trustStore = KeyStore.getInstance(KeyStore.getDefaultType());
            trustStore.load(null, null);
            trustStore.setCertificateEntry("ca", caCert);

            TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
            tmf.init(trustStore);

            // 2. （可选）加载客户端证书（双向认证）
            KeyManager[] keyManagers = null;
            // if (clientCertResource != null) {
            //     KeyStore keyStore = KeyStore.getInstance("PKCS12");
            //     try (InputStream keyInput = clientCertResource.getInputStream()) {
            //         keyStore.load(keyInput, clientPassword.toCharArray());
            //     }
            //     KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
            //     kmf.init(keyStore, clientPassword.toCharArray());
            //     keyManagers = kmf.getKeyManagers();
            // }

            SSLContext sslContext = SSLContext.getInstance("TLSv1.2");
            sslContext.init(keyManagers, tmf.getTrustManagers(), null);
            return sslContext.getSocketFactory();
        }
    }
    /**
     * 关闭MQTT连接
     */
    public void close() throws MqttException {
        mqttClient.close();
        mqttClient.disconnect();
    }

    /**
     * 重连
     */
    public void reConnect(String topic) {
        try {
            mqttClient.reconnect();
            while (true) {
                if (mqttClient.isConnected()) {
                    sub(topic);
                    break;
                } else {
                    Thread.sleep(30000);
                    log.info("重连中，请稍后......");
                }
            }
            log.info("MQTT连接成功，订阅成功.topic={}", topic);
        } catch (MqttException | InterruptedException e) {
            log.error("重连失败,失败原因{}", e.getMessage());
        }
    }

    /**
     * 向某个主题发布消息 默认qos：1
     *
     * @param topic:发布的主题
     * @param msg：发布的消息
     */
    public void pub(String topic, String msg) throws MqttException {
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setPayload(msg.getBytes());
        try {
            MqttTopic mqttTopic = mqttClient.getTopic(topic);
            MqttDeliveryToken token = mqttTopic.publish(mqttMessage);
            token.waitForCompletion();
        } catch (MqttException e) {
            log.error("MQTT发布失败,失败原因{}", e.getMessage());
        }
    }

    /**
     * 向某个主题发布消息
     *
     * @param topic: 发布的主题
     * @param msg:   发布的消息
     * @param qos:   消息质量    Qos：0、1、2
     */
    public void pub(String topic, String msg, int qos) throws MqttException {
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setQos(qos);
        mqttMessage.setPayload(msg.getBytes());
        MqttTopic mqttTopic = mqttClient.getTopic(topic);
        MqttDeliveryToken token = mqttTopic.publish(mqttMessage);
        token.waitForCompletion();
    }

    /**
     * 订阅某一个主题 ，此方法默认的的Qos等级为：1
     *
     * @param topic 主题
     */
    public void sub(String topic) throws MqttException {
        mqttClient.subscribe(topic);
    }

    /**
     * 订阅某一个主题，可携带Qos
     *
     * @param topic 所要订阅的主题
     * @param qos   消息质量：0、1、2
     */
    public void sub(String topic, int qos) throws MqttException {
        log.info("订阅的主题是 : " + topic);
        log.info("订阅的Qos是 : " + qos);
        mqttClient.subscribe(topic, qos);
    }

    /**
     * 取消某个主题发布消息
     */
    public void unSub(String topic) {
        log.info("取消订阅的主题是 : " + topic);
        try {
            mqttClient.unsubscribe(topic);
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }


}